package com.chinamobile.iot.handler.sub;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by szl on 2017/8/2.
 * this mqtt client is used to connect to other brokers to send subscription message
 */
@Component
@ConditionalOnExpression("${mqtt.distribute}")
public class NettyMqttClient {

    private Logger logger = LoggerFactory.getLogger(NettyMqttClient.class);
    private EventLoopGroup workerGroup = new NioEventLoopGroup(5);

    public boolean connect(String host, int port, String clientId, String userName, String password, boolean cleanSession, int keepAliveSeconds) {
        try {
            Bootstrap bootstrap = new Bootstrap(); // (1)
            bootstrap.group(workerGroup); // (2)
            bootstrap.channel(NioSocketChannel.class); // (3)
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true); // (4)
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, 10));
                    pipeline.addAfter("idleStateHandler", "idleEventHandler", new ClientTimeoutHandler());
                    pipeline.addLast("decoder", new MqttDecoder());
                    pipeline.addLast("encoder", MqttEncoder.INSTANCE);
                    pipeline.addLast("handler", new ClientMqttHandler(clientId, userName, password, cleanSession, keepAliveSeconds));
                }
            });

            // Start the client.
            ChannelFuture f = bootstrap.connect(host, port).sync(); // (5)
            f.sync();
            logger.info("connect to {}:{} successed", host, port);
        } catch (Exception e) {
            logger.warn("connect to {}:{} failed, message:{}", host, port, e.getMessage());
            return false;
        }
        return true;
    }

    public void destroy() {
        logger.info("NettyMqttClient::destroy() is called");
        try {
            workerGroup.shutdownGracefully();
        } catch (Exception e) {
            logger.warn("NettyMqttClient::destroy failed. {}", e.getMessage());
        }
    }

    public boolean isConnected(String host, int port) {
        return ClientMqttHandler.isConnected(host, port);
    }

    public boolean publish(String host, int port, String topic, String message, int qos) {
        return ClientMqttHandler.publish(host, port, topic, message, false, qos);
    }

    public boolean subscribe(String host, int port, String filter, int qos) {
        return ClientMqttHandler.subscribe(host, port, filter, qos);
    }

    public void close(String host, int port) {
        ClientMqttHandler.close(host, port);
    }

    private static class ConnectDiscr {
        private Channel channel;
        private AtomicInteger messageId = new AtomicInteger(1);

        public Channel getChannel() {
            return channel;
        }

        public void setChannel(Channel channel) {
            this.channel = channel;
        }

        public AtomicInteger getMessageId() {
            return messageId;
        }

        public void setMessageId(AtomicInteger messageId) {
            this.messageId = messageId;
        }

        public int nextMessageId() {
            int msgId = messageId.get();
            if (msgId >= 60000) {
                while (true) {
                    msgId = messageId.get();
                    if (msgId > 60000) {
                        if (messageId.compareAndSet(msgId, msgId - 60000)) {
                            break;
                        }
                    } else {
                        break;
                    }
                }
            }
            return messageId.incrementAndGet();
        }
    }

    public static class ClientTimeoutHandler extends ChannelDuplexHandler {

        private static Logger logger = LoggerFactory.getLogger(ClientTimeoutHandler.class);

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState e = ((IdleStateEvent) evt).state();
                if (e == IdleState.ALL_IDLE) {
                    logger.debug("no data up and down, begin send pingreq message to the server");
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
                    MqttMessage message = new MqttMessage(fixedHeader);
                    ctx.channel().writeAndFlush(message);
                }
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }

    public static class ClientMqttHandler extends ChannelInboundHandlerAdapter {

        private final static ConcurrentHashMap<String, ConnectDiscr> mapChannel = new ConcurrentHashMap<>(16);
        private static Logger logger = LoggerFactory.getLogger(ClientMqttHandler.class);
        private String clientId;
        private String userName;
        private String password;
        private boolean cleanSession;
        private int keepAliveSeconds;

        private ClientMqttHandler(String clientId, String userName, String password, boolean cleanSession, int keepAliveSeconds) {
            this.clientId = clientId;
            this.userName = userName;
            this.password = password;
            this.cleanSession = cleanSession;
            this.keepAliveSeconds = keepAliveSeconds;
        }


        public static boolean isConnected(String host, int port) {
            String key = "tcp://" + host + ":" + port;
            ConnectDiscr conn = mapChannel.get(key);
            if (conn != null) {
                return true;
            }
            return false;
        }

        public static void close(String host, int port) {
            String key = "tcp://" + host + ":" + port;
            ConnectDiscr conn = mapChannel.get(key);
            if (conn != null) {
                if (conn.channel != null) {
                    conn.channel.close();
                }
            }
        }

        public static boolean publish(String host, int port, String topic, String message, boolean isDup, int qos) {
            String key = "tcp://" + host + ":" + port;
            ConnectDiscr conn = mapChannel.get(key);
            if (conn != null) {
                if (conn.channel != null) {
                    ByteBuf payload = Unpooled.wrappedBuffer(message.getBytes());
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, isDup, MqttQoS.valueOf(qos), false, 0);
                    MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topic, conn.nextMessageId());
                    MqttPublishMessage message1 = new MqttPublishMessage(fixedHeader, variableHeader, payload);
                    if (conn.channel.isWritable()) {
                        conn.channel.writeAndFlush(message1);
                        return true;
                    } else {
                        return false;
                    }
                }
            }
            return false;
        }

        public static boolean subscribe(String host, int port, String filter, int qos) {
            String key = "tcp://" + host + ":" + port;
            ConnectDiscr conn = mapChannel.get(key);
            if (conn != null) {
                if (conn.channel != null) {
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.valueOf(qos), false, 0);
                    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(conn.nextMessageId());
                    List<MqttTopicSubscription> filters = new LinkedList<>();
                    MqttTopicSubscription subscription = new MqttTopicSubscription(filter, MqttQoS.valueOf(qos));
                    filters.add(subscription);
                    MqttSubscribePayload payload = new MqttSubscribePayload(filters);
                    MqttSubscribeMessage message1 = new MqttSubscribeMessage(fixedHeader, variableHeader, payload);
                    conn.channel.writeAndFlush(message1);
                    return true;
                }
            }
            return false;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
            String host = insocket.getAddress().getHostAddress();
            int port = insocket.getPort();
            String key = "tcp://" + host + ":" + port;
            logger.info("NettyMqttClient::channelActive, key = {}", key);

            ConnectDiscr connectDiscr = new ConnectDiscr();
            connectDiscr.setChannel(ctx.channel());
            mapChannel.put(key, connectDiscr);

            MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
            MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader("MQTT", MqttVersion.MQTT_3_1_1.protocolLevel()
                    , userName != null, password != null, false, 0, false, cleanSession, keepAliveSeconds);
            MqttConnectPayload payload = new MqttConnectPayload(clientId, null, null, userName, password);
            MqttConnectMessage connectMessage = new MqttConnectMessage(header, variableHeader, payload);
            ctx.channel().writeAndFlush(connectMessage);
            setIdleTime(ctx.channel().pipeline(), keepAliveSeconds / 2);
        }

        private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
            if (pipeline.names().contains("idleStateHandler")) {
                pipeline.remove("idleStateHandler");
            }
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
            String host = insocket.getHostName();
            int port = insocket.getPort();
            String key = "tcp://" + host + ":" + port;
            mapChannel.remove(key);
            ctx.channel().close();
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            super.handlerAdded(ctx);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
            MqttMessage msg = (MqttMessage) message;
            MqttMessageType messageType = msg.fixedHeader().messageType();
            logger.debug("Processing MQTT message, type={}", messageType);
            try {
                switch (messageType) {
                    case CONNECT: {
                    }
                    break;
                    case CONNACK: {

                    }
                    break;
                    case SUBSCRIBE: {

                    }
                    break;
                    case UNSUBSCRIBE: {

                    }
                    break;
                    case PUBLISH: {

                    }
                    break;
                    case PUBREC: {

                    }
                    break;
                    case PUBCOMP: {
                    }
                    break;
                    case PUBREL: {
                    }
                    break;
                    case DISCONNECT: {
                        ctx.close();
                    }
                    break;
                    case PUBACK: {
                    }
                    break;
                    case PINGREQ: {
                    }
                    break;
                    case PINGRESP: {

                    }
                    break;
                    default: {
                        logger.error("Unkonwn MessageType:{}", messageType);
                    }
                    break;
                }
            } catch (Throwable ex) {
                logger.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex);
                ctx.fireExceptionCaught(ex);
                ctx.close();
            } finally {
                ReferenceCountUtil.release(msg);
            }
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }


}
